package org.enhance.local.tq.task;

import lombok.extern.slf4j.Slf4j;
import org.enhance.local.tq.factory.TaskQueueClientFactory;
import org.enhance.local.tq.message.TaskObj;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

@Slf4j
public class TqThread<T extends ITqHandler> extends Thread {

    private T tqHandler;
    private LinkedBlockingQueue<TaskObj> taskQueue;

    public TqThread(T tqHandler) {
        this.tqHandler = tqHandler;
        taskQueue = TaskQueueClientFactory.getTaskObjQueue(tqHandler.getTaskType());
    }

    @Override
    public void run() {
        while (!TaskQueueClientFactory.clientFactoryPoolStoped) {
            try {
                TaskObj task = taskQueue.take();
                if (task != null) {
                    // load
                    List<TaskObj> taskList = new ArrayList<>();
                    taskList.add(task);

                    List<TaskObj> otherTaskList = new ArrayList<>();
                    int drainToNum = taskQueue.drainTo(otherTaskList, 100);
                    if (drainToNum > 0) {
                        taskList.addAll(otherTaskList);
                    }
                    tqHandler.run(taskList);
                }
            } catch (Exception e) {
                if (!TaskQueueClientFactory.clientFactoryPoolStoped) {
                    log.error(e.getMessage(), e);
                }
            }

            List<TaskObj> otherTaskList = new ArrayList<>();
            int drainToNum = taskQueue.drainTo(otherTaskList);
            if (drainToNum> 0) {
                tqHandler.run(otherTaskList);
            }
        }
    }
}
